package com.atguigu.gmall.realtime.tutorial.sql;

import com.atguigu.gmall.realtime.app.udf.KeywordUDTF;
import com.atguigu.gmall.realtime.tutorial.beans.Sensor;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class SQLTEST_Kafka {
    public static void main(String[] args)  throws Exception{
//TODO 0 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //CK设置

        //TODO 1 定义Table流环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,settings);

        // TODO 2 使用FlinkSQL 定义table 并且通过sql查询计算指标
        String sql =
                "CREATE TABLE sensor(id STRING,money BIGINT,temperature DOUBLE)    WITH (   " + MyKafkaUtil.getKafkaDDLTest("sensor","sensor_test","localhost:9092")
                + ")";
        System.out.println(sql);
        tableEnv.executeSql(sql);

//        Table table = tableEnv.sqlQuery("select id,money, temperature from sensor");
//        DataStream<Sensor> sensorReadingDataStream = tableEnv.toAppendStream(table, Sensor.class);
//        sensorReadingDataStream.print();

        //TODO   execute 方式只能用在kafka这样的流式中，filesystem会失败报错
//        TableResult execute = tableEnv.sqlQuery("select id,money, temperature from sensor  ").execute();
//        execute.print();


        // TODO 2 注册自定义函数
//        tableEnv.createTemporarySystemFunction("sensorName", SensorNameUDF.class);
//
//        TableResult new_name = tableEnv.sqlQuery("select sensorName(id) new_name from sensor ").execute();
//        new_name.print();


        // TODO 3 注册自动以UDTF函数
        tableEnv.createTemporarySystemFunction("sensorName", SensorNameUDTF.class);
        TableResult new_name = tableEnv.sqlQuery("select id,new_name,new_length from sensor, LATERAL TABLE(sensorName(id)) as T(new_name,new_length)").execute();
        new_name.print();


        env.execute();

    }
}
